package net.flyinggroup.kafka.app;


import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.binder.PollableMessageSource;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;

@SpringBootApplication
@EnableBinding(Blog2Application.PolledProcessor.class)
@Slf4j
public class Blog2Application {

    public static void main(String[] args) {
        SpringApplication.run(Blog2Application.class, args);
    }

    @Bean
    @SuppressWarnings("InfiniteLoopStatement")
    public ApplicationRunner runner(PollableMessageSource source, MessageChannel dest) {
        return args -> {
            while (true) {
                boolean result = source.poll(m -> {
                    String payload = new String((byte[]) m.getPayload());
                    log.info("Received: " + payload);
                    dest.send(MessageBuilder.withPayload(payload.toUpperCase()).copyHeaders(m.getHeaders()).build());
                });
                if (result) {
                    log.info("Processed a message");
                } else {
                    log.info("Nothing to do");
                }
                Thread.sleep(1000);
            }
        };
    }

    public interface PolledProcessor {
        @Input
        PollableMessageSource source();

        @Output
        MessageChannel dest();
    }
}